package cn.itcast.flink.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.TimeUnit;


/**
 * @author lilulu
 */
public class StreamSourceMysqlDemoVersion2 {

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }

    /**
     * 自定义数据源，从MySQL表中加载数据，并且实时增量加载，每隔5秒钟加载一次
     */
    public static class MysqlSource extends RichParallelSourceFunction<Student> {

        private boolean isRunning = true;

        private Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;

        /**
         * todo: 初始化方法，在调用run方法之前，调用的方法，每个实例对象，仅仅调用1次，比如获取连接
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            Class.forName("com.mysql.jdbc.Driver");
            connection = DriverManager.getConnection("jdbc:mysql://node1:3306/?useSSL=false", "root", "123456");
            preparedStatement = connection.prepareStatement("select * from db_flink.t_student");
        }


        public void run(SourceContext<Student> sourceContext) throws Exception {
            while (isRunning) {
                /*
                    todo 从Mysql数据库加载数据，使用jdbc方式读取数据
                 */
                // step4. 执行操作
                 resultSet = preparedStatement.executeQuery();
                while (resultSet.next()) {
                    int id = resultSet.getInt("id");
                    String name = resultSet.getString("name");
                    int age = resultSet.getInt("age");
                    Student student = new Student(id,name,age);
                    sourceContext.collect(student);
                }
                TimeUnit.SECONDS.sleep(5);
            }
        }

        public void cancel() {
            isRunning = false;
        }

        //关闭资源
        @Override
        public void close() throws Exception {
            if (resultSet != null) {
                resultSet.close();
            }
            if (preparedStatement != null) {
                preparedStatement.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<Student> studentDataStreamSource = env.addSource(new MysqlSource());
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        studentDataStreamSource.printToErr();
        // 5. 触发执行-execute
        env.execute("StreamSourceMysqlDemo");
    }
}